package cn.azzhu.day04;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author azzhu
 * @create 2020-09-19 14:10:43
 */
public class DataToActivityBeanFunction extends RichMapFunction<String,ActivityBean> {
    private Connection connection = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //创建MySQL连接,实现创建，只创建一次
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?characterEncoding=utf-8","root","root");
    }

    @Override
    public ActivityBean map(String line) throws Exception {
        final String[] fields = line.split(",");
        String uid = fields[0];
        String aid = fields[1];

        //根据aid作为查询条件查询出name,关联查询最好用redis，比较快
        final PreparedStatement ps = connection.prepareStatement("select name from t_activities where id=?");

        ps.setString(1,aid);
        final ResultSet rs = ps.executeQuery();
        String activityName = null;
        while (rs.next()) {
            activityName = rs.getString(1);
        }

       // rs.close();
        ps.close();

        String time = fields[2];
        final int eventType = Integer.parseInt(fields[3]);
        String province = fields[4];
        return ActivityBean.of(uid,aid,activityName,time,eventType,0L,0L,province);
    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }
}
